package protobuf

import (
	"encoding/binary"
	"errors"
	"gitee.com/jmesyan/impetus/log"
	"gitee.com/jmesyan/impetus/network"
	"gitee.com/jmesyan/impetus/protocol"
	"gitee.com/jmesyan/impetus/protocol/pomelo/message"
	"gitee.com/jmesyan/impetus/protocol/protobuf/pkg"
	"gitee.com/jmesyan/impetus/rpc"
	"gitee.com/jmesyan/impetus/service/codecService"
	"github.com/smallnest/rpcx/client"
	"net"
	"reflect"
)

type agent struct {
	conn      network.Conn
	connector *Connector
	session   *rpc.Session
	userData  interface{}
	agentHandler	*agentHandler
	chanRet		chan *client.Call
	writeChan 	chan *[]byte
}

const (
	msgRouteCompressMask = 0x01
	msgTypeMask          = 0x07
	msgRouteLengthMask   = 0xFF
	msgHeadLength        = 0x02
)

type MsgType byte

// Message types
const (
	MsgRequest  MsgType = 0x00
	MsgNotify        = 0x01
	MsgResponse      = 0x02
	MsgPush          = 0x03
)

type PkgType byte

const (
	_ PkgType = iota
	// Handshake represents a handshake: request(client) <====> handshake response(server)
	PkgHandshake = 0x01

	// HandshakeAck represents a handshake ack from client to server
	PkgHandshakeAck = 0x02

	// Heartbeat represents a heartbeat
	PkgHeartbeat = 0x03

	// Data represents a common data packet
	PkgData = 0x04

	// Kick represents a kick off packet
	PkgKick = 0x05 // disconnect message from server
)

const (
	HeadLength    = 4
	MaxPacketSize = 64 * 1024
)

type Message struct {
	Type       MsgType   // message type
	ID         uint64 // unique id, zero while notify mode
	Route      string // route for locating service
	Data       []byte // payload
	Payload    []byte
	compressed bool   // is message compressed
}

var (
	routes = make(map[string]uint16) // route map to code
	codes  = make(map[uint16]string) // code map to route
)

var (
	ErrWrongMessageType  = errors.New("wrong message type")
	ErrInvalidMessage    = errors.New("invalid message")
	ErrRouteInfoNotFound = errors.New("route info not found in dictionary")
	ErrWrongMessage      = errors.New("wrong message")
	ErrWrongPacketType = errors.New("wrong packet type")
)


func routable(t MsgType) bool {
	return t == MsgRequest || t == MsgNotify || t == MsgPush
}

func NewAgent(conn network.Conn, connector *Connector) *agent{
	a := &agent{
		conn:      conn,
		connector: connector,
		session:   rpc.NewSession(connector.nodeId),
		userData:  nil,
		chanRet: make(chan *client.Call, 100),
		writeChan: make(chan *[]byte, 100),
	}
	a.agentHandler = NewAgentHandler(a)
	return a
}

// Decode packet data length byte to int(Big end)
func bytesToInt(b []byte) int {
	result := 0
	for _, v := range b {
		result = result<<8 + int(v)
	}
	return result
}

// Encode packet data length to bytes(Big end)
func intToBytes(n int) []byte {
	buf := make([]byte, 3)
	buf[0] = byte((n >> 16) & 0xFF)
	buf[1] = byte((n >> 8) & 0xFF)
	buf[2] = byte(n & 0xFF)
	return buf
}


func (a *agent) Run() {
	go func() {
		defer a.conn.Close()
		for {
			select {
			case ri := <-a.chanRet:
				if ri.Error != nil {
					log.Error("failed to call: %v", ri.Error)
				} else {
					args := ri.Args.(*rpc.Args)
					if a.connector.handlerFilter != nil {
						a.connector.handlerFilter.After(ri.ServicePath + "." + ri.ServiceMethod, ri)
					}

					a.WriteResponse(args.MsgId, ri.Reply)
				}
			case b := <-a.writeChan:
				if b == nil {
					return
				}

				err := a.conn.WriteMessage(*b)
				protocol.FreePoolBuffer(b)
				if err != nil {
					log.Error("ws WriteMessage: %s", err.Error())
					return
				}
			}
		}
	}()

	switch a.conn.(type) {
	case *network.WSConn:
		//for {
		//	buffer := protocol.GetPoolMsg()
		//	err := a.conn.ReadMsg(buffer)
		//	if err != nil {
		//		log.Debug("read message: %v", err)
		//		break
		//	}
		//
		//	_, pkgType, body := pkg.Decode(buffer.Bytes())
		//	a.agentHandler.Handle(pkgType, body)
		//	buffer.Reset()
		//	protocol.FreePoolMsg(buffer)
		//}
		break
	case *network.TCPConn:
		for {
			headBuffer := protocol.GetUint32PoolData()

			// read len
			if _, err := a.conn.Read(*headBuffer); err != nil {
				log.Error("read headBuffer err:%v",err)
				protocol.PutUint32PoolData(headBuffer)
				break
			}
			pkgType := (*headBuffer)[0]
			// parse len
			//var msgLen uint32
			//if pkg.GetByteOrder() {
			//	msgLen = binary.LittleEndian.Uint32((*headBuffer)[1:])
			//} else {
			//	msgLen = binary.BigEndian.Uint32((*headBuffer)[1:])
			//}
			msgLen := bytesToInt((*headBuffer)[1:])


			// check len
			//if msgLen > p.maxMsgLen {
			//	return nil, errors.New("message too long")
			//} else if msgLen < p.minMsgLen {
			//	return nil, errors.New("message too short")
			//}
			m := new(Message)
			payloadBuffer := protocol.GetPoolBuffer(msgLen)
			if msgLen > 0{
				// data
				protocol.PutUint32PoolData(headBuffer)
				if _, err := a.conn.Read(*payloadBuffer); err != nil {
					log.Error("read payloadBuffer err:%v",err)
					protocol.FreePoolBuffer(payloadBuffer)
					break
				}
				//pkgType, body := pkg.Decode(*payloadBuffer)
				data := *payloadBuffer
				m.Payload = data
				flag := data[0]
				offset := 1

				m.Type = MsgType((flag>>1)&msgTypeMask)
				if m.Type == MsgRequest || m.Type ==MsgResponse{
					id := uint64(0)
					// little end byte order
					// WARNING: must can be stored in 64 bits integer
					// variant length encode
					for i := offset; i < len(data); i++ {
						b := data[i]
						id += uint64(b&0x7F) << uint64(7*(i-offset))
						if b < 128 {
							offset = i + 1
							break
						}
					}
					m.ID = id
				}
				if offset >= len(data) {
					break
				}

				if routable(m.Type) {
					if flag&msgRouteCompressMask == 1 {
						m.compressed = true
						code := binary.BigEndian.Uint16(data[offset:(offset + 2)])
						route, ok := codes[code]
						if !ok {
							break
						}
						m.Route = route
						offset += 2
					} else {
						m.compressed = false
						rl := data[offset]
						offset++
						if offset+int(rl) >= len(data) {
							break
						}
						m.Route = string(data[offset:(offset + int(rl))])
						offset += int(rl)
					}
				}

				if offset >= len(data) {
					break
				}
				m.Data = data[offset:]
			}
			a.agentHandler.Handle(pkgType, m)
			protocol.FreePoolBuffer(payloadBuffer)

		}
		break
	}

	close(a.writeChan)
}

func (a *agent) OnClose() {
	if a.agentHandler.timerHandler != nil {
		a.connector.timers.ClearTimeout(a.agentHandler.timerHandler)
	}
	a.connector.connection.OnDisconnect(a.session)
	a.connector.sessions.DelSession(a)
}


func invalidType(t MsgType) bool {
	return t < MsgRequest || t > MsgPush

}

// Encode marshals message to binary format. Different message types is corresponding to
// different message header, message types is identified by 2-4 bit of flag field. The
// relationship between message types and message header is presented as follows:
// ------------------------------------------
// |   type   |  flag  |       other        |
// |----------|--------|--------------------|
// | request  |----000-|<message id>|<route>|
// | notify   |----001-|<route>             |
// | response |----010-|<message id>        |
// | push     |----011-|<route>             |
// ------------------------------------------
// The figure above indicates that the bit does not affect the type of message.
// See ref: https://github.com/lonnng/nano/blob/master/docs/communication_protocol.md
func EncodeMsg(m *Message) ([]byte, error) {
	if invalidType(m.Type) {
		return nil, ErrWrongMessageType
	}

	buf := make([]byte, 0)
	flag := byte(m.Type) << 1

	code, compressed := routes[m.Route]
	if compressed {
		flag |= msgRouteCompressMask
	}
	buf = append(buf, flag)

	if m.Type == MsgRequest || m.Type == MsgResponse {
		n := m.ID
		// variant length encode
		for {
			b := byte(n % 128)
			n >>= 7
			if n != 0 {
				buf = append(buf, b+128)
			} else {
				buf = append(buf, b)
				break
			}
		}
	}

	if routable(m.Type) {
		if compressed {
			buf = append(buf, byte((code>>8)&0xFF))
			buf = append(buf, byte(code&0xFF))
		} else {
			buf = append(buf, byte(len(m.Route)))
			buf = append(buf, []byte(m.Route)...)
		}
	}

	buf = append(buf, m.Data...)
	return buf, nil
}

func EncodePkg(typ PkgType, data []byte) ([]byte, error) {
	if typ < PkgHandshake || typ > PkgKick {
		return nil, ErrWrongPacketType
	}
	length := len(data)
	buf := make([]byte,length+HeadLength)
	buf[0] = byte(typ)

	copy(buf[1:HeadLength], intToBytes(length))
	copy(buf[HeadLength:], data)
	return buf, nil
}

func (a *agent) WriteResponse(msgId int, msg interface{}) {
	_codec := codecService.GetCodecService()
	if _codec != nil {
		data, err := _codec.Marshal(msg)
		if err != nil {
			log.Error("marshal message %v error: %v", reflect.TypeOf(msg), err)
			return
		}
		m := &Message{
			Type:       MsgResponse,
			ID:         uint64(msgId),
			Route:      "",
			Data:       data,
			Payload:    nil,
			compressed: false,
		}
		//buffer := pkg.Encode(msgId, data)
		buffer,err := EncodeMsg(m)
		if err != nil {
			log.Error("encode response EncodeMsg %v error: %v", reflect.TypeOf(msg), err)
			return
		}
		buf, err := EncodePkg(PkgData,buffer)
		if err != nil {
			log.Error("encode response EncodePkg %v error: %v", reflect.TypeOf(msg), err)
			return
		}
		err = a.conn.WriteMessage(buf)
		//protocol.FreePoolBuffer(buffer)
		if err != nil {
			log.Error("write message %v error: %v", reflect.TypeOf(msg), err)
		}
	}
}

// Write to channel. Make sure buffer from protocol.GetPoolBuffer()
func (a *agent) Write(data *[]byte) {
	a.writeChan <- data
}

func (a *agent) LocalAddr() net.Addr {
	return a.conn.LocalAddr()
}

func (a *agent) RemoteAddr() net.Addr {
	return a.conn.RemoteAddr()
}

func (a *agent) Close() {
	a.conn.Close()
}

func (a *agent) UserData() interface{} {
	return a.userData
}

func (a *agent) SetUserData(data interface{}) {
	a.userData = data
}

func (a *agent) GetSession() *rpc.Session {
	return a.session
}

func (a *agent) PushMessage(routeId uint32, data []byte) {
	buffer := message.Encode(0, message.TYPE_PUSH, uint16(routeId), data)
	a.Write(pkg.Encode(pkg.TYPE_DATA, buffer))
}

func (a *agent) KickMessage(reason string) {
	ret := map[string]string{
		"reason": reason,
	}
	buffer,_ := codecService.GetCodecService().Marshal(ret)
	a.Write(pkg.Encode(pkg.TYPE_KICK, buffer))
}